RNN with Keras, scikit-learn

데이터 전처리

from keras.models import Sequential
from keras.layers import SimpleRNN, Dense
from sklearn.preprocessing import MinMaxScaler
from sklearn.metrics import mean_squared_error
from sklearn.model_selection import train_test_split

dataframe = read_csv('', usecols=[3], engine='python', skipfooter=3)
dataset = dataframe.values
dataset = dataset.astype('float32')

scaler = MinMaxScaler(feature_range=(0, 1))
Dataset = scaler.fit_transform(dataset) #Normalization
train_data, test_data = train_test_split(Dataset, test_size=0.2, shuffle=False)

from sklearn.model_selection import train_test_split: 데이터를 훈련 데이터와 검증 데이터로 나눔.
- test_size를 통해 검증 데이터 비율을 정함.
- shuffle: 순서가 중요한 시계열 데이터에는 사용하지 않음.
from sklearn.preprocessing import MinMaxScaler: 데이터의 Nomalization. feature_range을 통해 Nomarlization 범위를 설정한다.
from sklearn.metrics import mean_squared_error: 결과의 정확도를 계산하기 위한 함수, 실제 값과 예측 값의 차이를 사용하여 Loss 구한다.

시계열 데이터

def create_dataset(dataset, look_back):
    x_data = []
    y_data = []
    for i in range(len(dataset)-look_back-1):
        data = dataset[i:(i+look_back), 0]
        x_data.append(data)
        y_data.append(dataset[i + look_back, 0])
  return np.array(x_data), np.array(y_data)

look_back = 3
x_train, y_train = create_dataset(train_data, look_back)
x_test, y_test = create_dataset(test_data, look_back)

X_train = np.reshape(x_train, (x_train.shape[0], 1, x_train.shape[1]))
X_test = np.reshape(x_test, (x_test.shape[0], 1, x_test.shape[1]))
# 85 x 3 => 85 x 1 x 3, (1 x 3) 크기의 배열을 85개로 나누어 넣기 위함.

RNN 모델 설계

model = Sequential()
model.add(SimpleRNN(3, input_shape=(1, look_back)))
model.add(Dense(1, activation="linear")) # 연속된 데이터 이후의 값으로, 예측값. 1개의 노드만 설정(출력층)

model.compile(loss='mse', optimizer='adam')
model.fit(X_train, y_train, epochs=100, batch_size=1, verbose=1)

SimpleRNN: (Node 갯수, input_shape)
mse: 평균 제곱 오차(mean_squared_error)

데이터 역정규화

trainPredict = model.predict(X_train)
testPredict = model.predict(X_test)
TrainPredict = scaler.inverse_transform(trainPredict)
Y_train = scaler.inverse_transform([y_train])
TestPredict = scaler.inverse_transform(testPredict)
Y_test = scaler.inverse_transform([y_test])

오차 구하기

trainScore = math.sqrt(mean_squared_error(Y_train[0], TrainPredict[:,0]))
testScore = math.sqrt(mean_squared_error(Y_test[0], TestPredict[:,0]))